package com.weishao.example;

import java.util.LinkedList;
import java.util.List;
import javax.sql.DataSource;
import org.apache.commons.dbcp2.BasicDataSource;
import com.weishao.common.util.JdbcMetaDataUtils;
import com.weishao.dbchange.ChangeCaculatorService;
import com.weishao.dbchange.IDatabaseChangeCaculator;
import com.weishao.dbchange.IDatabaseRowHandler;
import com.weishao.dbchange.RecordChangeType;
import com.weishao.dbchange.pojo.TaskParamBean;
import com.weishao.dbsynch.DatabaseSynchronizeFactory;
import com.weishao.dbsynch.IDatabaseSynchronize;
import com.zaxxer.hikari.HikariDataSource;

public class DbsynchExampleApplication {

	private static int BATCH_SIZE = 1000;

	public static void testSynchData(DataSource sds, DataSource tds) {
		/**
		 * 将old里的数据同步为new里的数据
		 */
		TaskParamBean.TaskParamBeanBuilder taskBuilder = TaskParamBean.builder();
		taskBuilder.oldDataSource(tds);
		taskBuilder.oldSchemaName("test");
		taskBuilder.oldTableName("new_user");
		taskBuilder.newDataSource(sds);
		taskBuilder.newSchemaName("tang");
		taskBuilder.newTableName("new_user");

		TaskParamBean task = taskBuilder.build();

		IDatabaseSynchronize synch = DatabaseSynchronizeFactory.createDatabaseWriter(tds);
		JdbcMetaDataUtils mdt = new JdbcMetaDataUtils(tds);
		List<String> pks = mdt.queryTablePrimaryKeys(task.getOldSchemaName(), task.getOldTableName());
		List<String> fieldNames = mdt.queryTableColumnName(task.getOldSchemaName(), task.getOldTableName());
		synch.prepare(task.getOldSchemaName(), task.getOldTableName(), fieldNames, pks);

		IDatabaseChangeCaculator changeCaculator = new ChangeCaculatorService();
		changeCaculator.setFetchSize(BATCH_SIZE);
		changeCaculator.setRecordIdentical(false);

		changeCaculator.executeCalculate(task, new IDatabaseRowHandler() {

			private long count = 0;
			private List<Object[]> cacheInsert = new LinkedList<Object[]>();
			private List<Object[]> cacheUpdate = new LinkedList<Object[]>();
			private List<Object[]> cacheDelete = new LinkedList<Object[]>();

			@Override
			public void handle(List<String> fields, Object[] record, RecordChangeType flag) {
				if (flag == RecordChangeType.VALUE_INSERT) {
					cacheInsert.add(record);
				} else if (flag == RecordChangeType.VALUE_CHANGED) {
					cacheUpdate.add(record);
				} else {
					cacheDelete.add(record);
				}

				count++;
				checkFull(fields);
			}

			private void checkFull(List<String> fields) {
				if (cacheInsert.size() >= BATCH_SIZE) {
					doInsert(fields);
				}

				if (cacheUpdate.size() >= BATCH_SIZE) {
					doUpdate(fields);
				}

				if (cacheDelete.size() >= BATCH_SIZE) {
					doDelete(fields);
				}
			}

			@Override
			public void destroy(List<String> fields) {
				if (cacheInsert.size() > 0) {
					doInsert(fields);
				}

				if (cacheUpdate.size() > 0) {
					doUpdate(fields);
				}

				if (cacheDelete.size() > 0) {
					doDelete(fields);
				}

				System.out.println("处理的数据记录总数为: " + count);
			}

			private void doInsert(List<String> fields) {
				long ret = synch.executeInsert(cacheInsert);
				System.out.println("处理Insert记录总数为:" + ret);
				cacheInsert.clear();
			}

			private void doUpdate(List<String> fields) {
				long ret = synch.executeUpdate(cacheUpdate);
				System.out.println("处理Update记录总数为:" + ret);
				cacheUpdate.clear();
			}

			private void doDelete(List<String> fields) {
				long ret = synch.executeDelete(cacheDelete);
				System.out.println("处理Delete记录总数为:" + ret);
				cacheDelete.clear();
			}

		});
	}

	public static void main(String[] args) {
		BasicDataSource ds1 = new BasicDataSource();
		ds1.setDriverClassName("org.postgresql.Driver");
		ds1.setUrl("jdbc:postgresql://172.17.207.151:5432/study");
		ds1.setUsername("study");
		ds1.setPassword("123456");
		ds1.setValidationQuery("select 1");
		ds1.setInitialSize(5);
		ds1.setMaxIdle(5);
		ds1.setMinIdle(1);

		HikariDataSource ds2 = new HikariDataSource();
		ds2.setJdbcUrl("jdbc:mysql://172.17.207.210:3306/test?useSSL=false");
		ds2.setDriverClassName("com.mysql.jdbc.Driver");
		ds2.setUsername("tangyibo");
		ds2.setPassword("tangyibo");
		ds2.setConnectionTestQuery("select 1");
		ds2.setMaximumPoolSize(5);
		ds2.setMinimumIdle(1);

		testSynchData(ds1, ds2);

		try {
			ds1.close();
			ds2.close();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

}
